package org.apache.rocketmq.store;
import com.sun.jna.Library;
import com.sun.jna.NativeLong;
import com.sun.jna.Pointer;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.store.util.LibC;
import sun.nio.ch.DirectBuffer;

import java.nio.ByteBuffer;
import java.util.Deque;
import java.util.concurrent.ConcurrentLinkedDeque;

/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/29
 * @方法描述：临时存储池，用于临时存储消息，当消息被存储到临时存储池后，会有后台线程定期把临时存储池中的消息提交到操作系统缓存页
 * 同步刷盘的时候，就会把操作系统缓存页中的数据刷新到本地硬盘
 */
public class TransientStorePool {
    private static final Logger log = LoggerFactory.getLogger(TransientStorePool.class);

    private final int poolSize;

    private final int fileSize;

    //存放内存缓存队列
    private final Deque<ByteBuffer> availableBuffers;
    //是否执行实际提交数据的操作，所谓提交数据就是把对外内存缓冲区的数据提交到操作系统缓存页中
    private volatile boolean isRealCommit = false;

    public TransientStorePool(int poolSize, int fileSize) {
        this.poolSize = poolSize;
        this.fileSize = fileSize;
        this.availableBuffers = new ConcurrentLinkedDeque<>();
    }

    public void init(){
        for (int i = 0; i < poolSize; i++){
            //申请对外内存，创建对外内存缓冲区对象
            ByteBuffer byteBuffer = ByteBuffer.allocateDirect(fileSize);
            //得到直接缓冲区的内存地址
            final long address = ((DirectBuffer) byteBuffer).address();
            //创建一个指向该地址的指针
            Pointer pointer = new Pointer(address);
            //对该内存上锁，保证这块缓冲区的数据不会被操作系统交换到硬盘
            //也就是不会被交换到Swap区
            LibC.INSTANCE.mlock(pointer, new NativeLong(fileSize));
            //把创建完毕的缓冲区添加到可用缓冲区队列中
            availableBuffers.offer(byteBuffer);
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：销毁缓冲池的方法
     */
    public void destroy() {
        for (ByteBuffer byteBuffer : availableBuffers) {
            //获取直接缓冲区的内存地址
            final long address = ((DirectBuffer) byteBuffer).address();
            //创建一个指向该地址的本地指针
            Pointer pointer = new Pointer(address);
            //对这些内存解锁，允许操作系统正常管理这部分内存，包括可能的交换操作
            LibC.INSTANCE.munlock(pointer, new NativeLong(fileSize));
        }
    }
    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：归还缓冲区给临时存储池
     */
    public void returnBuffer(ByteBuffer byteBuffer) {
        byteBuffer.position(0);
        byteBuffer.limit(fileSize);
        this.availableBuffers.offerFirst(byteBuffer);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：从临时存储池中借用一个缓冲区的方法
     */
    public ByteBuffer borrowBuffer() {
        ByteBuffer buffer = availableBuffers.pollFirst();
        //检查剩余可用缓冲区数量，如果低于池大小的40%，记录警告日志
        if (availableBuffers.size() < poolSize * 0.4) {
            log.warn("TransientStorePool only remain {} sheets.", availableBuffers.size());
        }
        return buffer;
    }


    public int availableBufferNums() {
        return availableBuffers.size();
    }


    //判断是否真的提交数据
    public boolean isRealCommit() {
        return isRealCommit;
    }


    public void setRealCommit(boolean realCommit) {
        isRealCommit = realCommit;
    }
}